Day 29: 完整專案整合與展示

🎉 集大成之作：打造你的 AI 助理作品集

經過 28 天的學習旅程，是時候將所有知識整合成一個完整、可展示的專案了。今天，我們將把 Gemini CLI 和 LangGraph 的所有精華融合在一起，打造一個令人印象深刻的 AI 助理系統。

🏗️ 專案架構總覽

一個完整的 AI 助理系統應該包含以下核心模組：

1. 智能對話引擎
整合 Gemini API，支援多輪對話、上下文理解、情感分析。使用 LangGraph 管理對話狀態，確保每次互動都能記住先前的內容。

2. 多功能工作流程
實現條件路由系統：根據使用者意圖自動分流到不同的處理節點——文件分析、程式碼生成、數據處理或一般問答。每個節點都是獨立的專家系統。

3. 外部整合能力
連接天氣 API、新聞 API、資料庫查詢等外部服務。透過 LangGraph 的工具調用機制，讓 AI 助理能夠取得即時資訊並執行實際操作。

4. 記憶與個性化
使用向量資料庫（如 ChromaDB）儲存長期記憶，結合 Redis 快取短期對話。系統能記住使用者偏好，提供個性化建議。

💻 核心技術整合

"""
完整 AI 助理系統 - 整合 Gemini CLI 與 LangGraph
展示 30 天學習成果的完整專案
"""

import os
import json
from datetime import datetime
from typing import TypedDict, Annotated, List, Dict, Any
from dataclasses import dataclass

# LangGraph 相關
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolExecutor
from langchain.tools import tool

# Gemini API（假設使用 google-generativeai）
import google.generativeai as genai


# ============================================
# 1. 狀態定義
# ============================================

class AgentState(TypedDict):
    """AI 助理的完整狀態"""
    messages: List[str]
    current_intent: str
    user_profile: Dict[str, Any]
    context: Dict[str, Any]
    tools_used: List[str]
    response: str
    metadata: Dict[str, Any]


# ============================================
# 2. Gemini 包裝器
# ============================================

class GeminiAssistant:
    """Gemini API 智能包裝器"""
    
    def __init__(self, api_key: str):
        genai.configure(api_key=api_key)
        self.model = genai.GenerativeModel('gemini-pro')
        self.chat_history = []
    
    def generate(self, prompt: str, context: Dict = None) -> str:
        """生成回應，支援上下文"""
        if context:
            prompt = f"背景資訊：{json.dumps(context, ensure_ascii=False)}\n\n{prompt}"
        
        response = self.model.generate_content(prompt)
        self.chat_history.append({
            "prompt": prompt,
            "response": response.text,
            "timestamp": datetime.now().isoformat()
        })
        
        return response.text
    
    def analyze_intent(self, message: str) -> str:
        """分析使用者意圖"""
        prompt = f"""
        分析以下使用者訊息的意圖，僅回覆意圖類型（單一詞）：
        - document_analysis（文件分析）
        - code_generation（程式碼生成）
        - data_query（數據查詢）
        - weather_query（天氣查詢）
        - chitchat（閒聊）
        
        訊息：{message}
        意圖：
        """
        
        return self.generate(prompt).strip().lower()


# ============================================
# 3. 工具定義
# ============================================

@tool
def analyze_document(file_path: str) -> Dict[str, Any]:
    """分析文件內容並提取重點"""
    # 實際應用中會讀取檔案內容
    return {
        "summary": "文件摘要內容",
        "key_points": ["重點1", "重點2", "重點3"],
        "word_count": 1500,
        "sentiment": "neutral"
    }

@tool
def generate_code(description: str, language: str = "python") -> str:
    """根據描述生成程式碼"""
    assistant = GeminiAssistant(os.getenv("GEMINI_API_KEY"))
    prompt = f"請生成 {language} 程式碼來實現：{description}\n只回傳程式碼，不要額外說明。"
    return assistant.generate(prompt)

@tool
def query_database(query: str) -> List[Dict]:
    """查詢資料庫"""
    # 模擬資料庫查詢
    return [
        {"id": 1, "name": "範例資料1", "value": 100},
        {"id": 2, "name": "範例資料2", "value": 200}
    ]

@tool
def get_weather(city: str) -> Dict[str, Any]:
    """取得天氣資訊"""
    # 實際應用中會調用天氣 API
    return {
        "city": city,
        "temperature": 25,
        "condition": "晴天",
        "humidity": 60
    }

# 工具執行器
tools = [analyze_document, generate_code, query_database, get_weather]
tool_executor = ToolExecutor(tools)


# ============================================
# 4. LangGraph 節點定義
# ============================================

def intent_classifier(state: AgentState) -> AgentState:
    """意圖分類節點"""
    assistant = GeminiAssistant(os.getenv("GEMINI_API_KEY"))
    last_message = state["messages"][-1]
    
    intent = assistant.analyze_intent(last_message)
    state["current_intent"] = intent
    state["metadata"]["intent_classified_at"] = datetime.now().isoformat()
    
    print(f"🎯 偵測到意圖: {intent}")
    return state


def document_handler(state: AgentState) -> AgentState:
    """文件處理節點"""
    print("📄 執行文件分析...")
    
    # 調用文件分析工具
    result = tool_executor.invoke({
        "tool": "analyze_document",
        "tool_input": {"file_path": "sample.pdf"}
    })
    
    assistant = GeminiAssistant(os.getenv("GEMINI_API_KEY"))
    response = assistant.generate(
        f"根據文件分析結果生成簡潔的回覆：{result}",
        context=state["context"]
    )
    
    state["response"] = response
    state["tools_used"].append("document_analysis")
    return state


def code_generator(state: AgentState) -> AgentState:
    """程式碼生成節點"""
    print("💻 生成程式碼...")
    
    last_message = state["messages"][-1]
    code = tool_executor.invoke({
        "tool": "generate_code",
        "tool_input": {"description": last_message, "language": "python"}
    })
    
    state["response"] = f"這是為您生成的程式碼：\n\n```python\n{code}\n```"
    state["tools_used"].append("code_generation")
    return state


def data_analyst(state: AgentState) -> AgentState:
    """數據分析節點"""
    print("📊 查詢數據...")
    
    query_result = tool_executor.invoke({
        "tool": "query_database",
        "tool_input": {"query": state["messages"][-1]}
    })
    
    assistant = GeminiAssistant(os.getenv("GEMINI_API_KEY"))
    response = assistant.generate(
        f"根據查詢結果生成分析報告：{query_result}"
    )
    
    state["response"] = response
    state["tools_used"].append("data_query")
    return state


def weather_assistant(state: AgentState) -> AgentState:
    """天氣助手節點"""
    print("🌤️ 查詢天氣...")
    
    # 從訊息中提取城市名稱
    assistant = GeminiAssistant(os.getenv("GEMINI_API_KEY"))
    city = assistant.generate(
        f"從這句話提取城市名稱，只回覆城市名：{state['messages'][-1]}"
    ).strip()
    
    weather = tool_executor.invoke({
        "tool": "get_weather",
        "tool_input": {"city": city}
    })
    
    state["response"] = f"{city}目前天氣：{weather['condition']}，溫度 {weather['temperature']}°C"
    state["tools_used"].append("weather_query")
    return state


def general_chat(state: AgentState) -> AgentState:
    """一般對話節點"""
    print("💬 進行一般對話...")
    
    assistant = GeminiAssistant(os.getenv("GEMINI_API_KEY"))
    response = assistant.generate(
        state["messages"][-1],
        context={
            "user_profile": state["user_profile"],
            "history": state["messages"][-5:]  # 最近5條訊息
        }
    )
    
    state["response"] = response
    return state


def personalize_response(state: AgentState) -> AgentState:
    """個性化回應節點"""
    print("✨ 個性化處理...")
    
    # 根據使用者偏好調整回應風格
    if state["user_profile"].get("style") == "formal":
        state["response"] = "您好，" + state["response"]
    elif state["user_profile"].get("style") == "casual":
        state["response"] = "嘿！" + state["response"]
    
    # 記錄使用者互動
    state["user_profile"]["interaction_count"] = \
        state["user_profile"].get("interaction_count", 0) + 1
    
    return state


# ============================================
# 5. 路由邏輯
# ============================================

def route_by_intent(state: AgentState) -> str:
    """根據意圖路由到不同節點"""
    intent = state["current_intent"]
    
    routing_map = {
        "document_analysis": "document_handler",
        "code_generation": "code_generator",
        "data_query": "data_analyst",
        "weather_query": "weather_assistant",
        "chitchat": "general_chat"
    }
    
    return routing_map.get(intent, "general_chat")


# ============================================
# 6. 構建工作流程圖
# ============================================

def create_ai_assistant_graph() -> StateGraph:
    """創建完整的 AI 助理工作流程圖"""
    
    workflow = StateGraph(AgentState)
    
    # 添加節點
    workflow.add_node("intent_classifier", intent_classifier)
    workflow.add_node("document_handler", document_handler)
    workflow.add_node("code_generator", code_generator)
    workflow.add_node("data_analyst", data_analyst)
    workflow.add_node("weather_assistant", weather_assistant)
    workflow.add_node("general_chat", general_chat)
    workflow.add_node("personalize", personalize_response)
    
    # 設置入口點
    workflow.set_entry_point("intent_classifier")
    
    # 添加條件路由
    workflow.add_conditional_edges(
        "intent_classifier",
        route_by_intent,
        {
            "document_handler": "document_handler",
            "code_generator": "code_generator",
            "data_analyst": "data_analyst",
            "weather_assistant": "weather_assistant",
            "general_chat": "general_chat"
        }
    )
    
    # 所有處理節點都導向個性化節點
    for node in ["document_handler", "code_generator", "data_analyst", 
                 "weather_assistant", "general_chat"]:
        workflow.add_edge(node, "personalize")
    
    # 個性化後結束
    workflow.add_edge("personalize", END)
    
    return workflow.compile()


# ============================================
# 7. 主應用程式
# ============================================

class AIAssistantApp:
    """完整的 AI 助理應用程式"""
    
    def __init__(self):
        self.graph = create_ai_assistant_graph()
        self.user_profiles = {}
    
    def get_or_create_profile(self, user_id: str) -> Dict[str, Any]:
        """取得或建立使用者檔案"""
        if user_id not in self.user_profiles:
            self.user_profiles[user_id] = {
                "id": user_id,
                "style": "casual",
                "preferences": {},
                "interaction_count": 0,
                "created_at": datetime.now().isoformat()
            }
        return self.user_profiles[user_id]
    
    def chat(self, user_id: str, message: str) -> str:
        """處理使用者訊息"""
        print(f"\n{'='*50}")
        print(f"👤 使用者 {user_id}: {message}")
        print(f"{'='*50}\n")
        
        # 準備初始狀態
        initial_state = {
            "messages": [message],
            "current_intent": "",
            "user_profile": self.get_or_create_profile(user_id),
            "context": {},
            "tools_used": [],
            "response": "",
            "metadata": {
                "timestamp": datetime.now().isoformat(),
                "user_id": user_id
            }
        }
        
        # 執行工作流程
        result = self.graph.invoke(initial_state)
        
        # 顯示結果
        print(f"\n{'='*50}")
        print(f"🤖 AI 助理: {result['response']}")
        print(f"📊 使用工具: {', '.join(result['tools_used']) if result['tools_used'] else '無'}")
        print(f"🎯 偵測意圖: {result['current_intent']}")
        print(f"{'='*50}\n")
        
        return result["response"]
    
    def get_statistics(self) -> Dict[str, Any]:
        """取得系統統計資訊"""
        return {
            "total_users": len(self.user_profiles),
            "total_interactions": sum(
                p.get("interaction_count", 0) 
                for p in self.user_profiles.values()
            ),
            "user_profiles": self.user_profiles
        }


# ============================================
# 8. 示範應用
# ============================================

def main():
    """主程式示範"""
    
    # 設置 API Key（實際使用時從環境變數讀取）
    os.environ["GEMINI_API_KEY"] = "your-api-key-here"
    
    # 初始化應用
    app = AIAssistantApp()
    
    print("🚀 完整 AI 助理系統啟動！")
    print("=" * 60)
    
    # 模擬不同場景的對話
    test_scenarios = [
        ("user_001", "幫我分析一下這份 PDF 文件的內容"),
        ("user_001", "寫一個 Python 函數來計算費波那契數列"),
        ("user_002", "台北今天天氣如何？"),
        ("user_002", "查詢我們資料庫中所有訂單"),
        ("user_001", "你好，今天過得怎麼樣？"),
    ]
    
    for user_id, message in test_scenarios:
        response = app.chat(user_id, message)
        # 實際應用中可以儲存對話歷史
    
    # 顯示統計資訊
    print("\n📊 系統統計資訊")
    print("=" * 60)
    stats = app.get_statistics()
    print(f"總使用者數: {stats['total_users']}")
    print(f"總互動次數: {stats['total_interactions']}")
    
    print("\n✅ 示範完成！")


if __name__ == "__main__":
    main()


# ============================================
# 9. 部署配置範例
# ============================================

"""
# Dockerfile
FROM python:3.10-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["python", "ai_assistant.py"]


# docker-compose.yml
version: '3.8'

services:
  ai-assistant:
    build: .
    ports:
      - "8000:8000"
    environment:
      - GEMINI_API_KEY=${GEMINI_API_KEY}
    volumes:
      - ./data:/app/data


# requirements.txt
google-generativeai>=0.3.0
langgraph>=0.0.20
langchain>=0.1.0
python-dotenv>=1.0.0
"""

🎨 專案展示重點

核心特色

1. 智能意圖識別
系統自動分析使用者意圖，將請求路由到最適合的處理節點。這是 LangGraph 條件路由的完美展現。

2. 模組化設計
每個功能獨立封裝成節點，易於擴展和維護。想新增功能？只需加入新節點和路由規則。

3. 狀態管理
完整追蹤對話歷史、使用者偏好、工具使用記錄。系統「記得」每個使用者的互動習慣。

4. 工具整合
無縫連接外部服務，從文件分析到天氣查詢，展現真實世界的實用性。

📱 展示策略

GitHub README 必備元素：

• 清晰的架構圖展示工作流程
• GIF 動畫展示實際運作
• 完整的安裝與使用說明
• 功能特色列表與技術棧

Demo 影片腳本：

1. 展示多輪對話能力
2. 演示不同意圖的智能路由
3. 顯示個性化回應效果
4. 展示系統監控儀表板

🚀 部署選項

• 本地部署：Docker Compose 一鍵啟動
• 雲端部署：支援 AWS、GCP、Azure
• 無伺服器：Cloud Functions 彈性擴展

💡 下一步優化方向

1. 加入語音輸入輸出
2. 整合更多第三方服務
3. 建立 Web UI 介面
4. 實現多語言支援
5. 加強安全性與隱私保護

這個完整專案不僅是學習成果的展示，更是你進入 AI 開發領域的敲門磚。明天 Day 30，我們將探討未來發展方向和持續學習資源，為這段旅程畫下完美句點！ 🎯